package avicit.bdp.dcs.task.service;

import avicit.bdp.common.database.ColumnMeta;
import avicit.bdp.common.dto.api.APIHttpDefinition;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.service.BaseDbService;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.datasource.service.TableService;
import avicit.bdp.dcs.detail.dto.TaskDataDetailDTO;
import avicit.bdp.dcs.detail.dto.TaskTableDetailDTO;
import avicit.bdp.dcs.detail.service.TaskDataDetailService;
import avicit.bdp.dcs.detail.service.TaskTableDetailService;
import avicit.bdp.dcs.job.dto.ApiDefinitionDto;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.job.service.DcsJobService;
import avicit.bdp.dcs.json.service.DataxJsonService;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.TableMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.Enable;
import avicit.bdp.dcs.tools.enums.ProcessMode;
import avicit.bdp.dcs.tools.enums.TaskDetailStatus;
import avicit.bdp.dcs.utils.DcsConstants;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.bdp.dcs.utils.api.enums.Key;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.exception.BusinessException;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @金航数码科技有限责任公司
 * @作者：liyb
 * @邮箱：liyb@avic-digital.com
 * @创建时间： 2020-09-08 15:36
 * @类说明： api任务服务类
 * @修改记录：
 */
@Service
@Transactional(rollbackFor = Exception.class)
public class ApiTaskService implements TaskInterface {

  private static final Logger logger = LoggerFactory.getLogger(ApiTaskService.class);

  @Autowired private TaskService taskService;
  @Autowired private DcsJobService jobService;
  @Autowired private DataxJsonService dataxJsonService;
  @Autowired private TableService tableService;
  @Autowired private TaskTableDetailService taskTableDetailService;
  @Autowired private TaskDataDetailService taskDataDetailService;
  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private BaseDbService baseDbService;
  @Autowired private ApiDefinitionService apiDefinitionService;
  @Autowired private DataSourceService dataSourceService;

  /** 查看任务 */
  @Override
  public TaskDTO getTaskById(JobDTO jobDTO, String srcTableName) throws Exception {
    // 1、查询任务
    TaskDTO query = new TaskDTO();
    query.setJobId(jobDTO.getId());
    query.setSrcTableName(srcTableName);
    query.setFlag(Enable.ZERO.getCode());
    TaskDTO task = taskService.selectOne(query);
    CheckParaUtils.checkDbObjectValidity(
        task,
        logger,
        String.format("没有查到数据，请确保jobId：%s,srcTableName:%s的任务是否存在", jobDTO.getId(), srcTableName));

    // 2、查询源表列字段
    List<ColumnInfo> srcColumnList = new ArrayList<>();
    task.setSrcColumnList(srcColumnList);

    // 3、获取列映射关系
    Map<String, String> columnMap = new HashMap<>();
    List<ColumnMappingDto> columnMappingDtoList =
        JSONUtils.toList(task.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        columnMap.put(columnMappingDto.getSrcColumnName(), columnMappingDto.getDstColumnName());
      }
    }

    // 4、查询目标表列字段
    List<ColumnInfo> lastColumnList = new ArrayList<>();
    List<ColumnInfo> dstColumnList =
        baseDbService.getColumnList(jobDTO.getDstDatasourceId(), task.getDstTableName());
    if (CollectionUtils.isNotEmpty(dstColumnList)) {
      for (ColumnInfo columnInfo : dstColumnList) {
        ColumnInfo dstColumnInfo = new ColumnInfo();
        String modifyColumnName = columnMap.get(columnInfo.getName());
        dstColumnInfo.setName(modifyColumnName == null ? columnInfo.getName() : modifyColumnName);
        dstColumnInfo.setName(dstColumnInfo.getName().toLowerCase());
        dstColumnInfo.setType(columnInfo.getType());
        dstColumnInfo.setLength(columnInfo.getLength());
        lastColumnList.add(dstColumnInfo);
      }
    } else {
      for (ColumnInfo columnInfo : task.getSrcColumnList()) {
        ColumnInfo dstColumnInfo = new ColumnInfo();
        String modifyColumnName = columnMap.get(columnInfo.getName());
        dstColumnInfo.setName(modifyColumnName == null ? columnInfo.getName() : modifyColumnName);
        dstColumnInfo.setType(columnInfo.getType());
        dstColumnInfo.setLength(columnInfo.getLength());
        lastColumnList.add(dstColumnInfo);
      }
    }

    lastColumnList =
        taskService.sortBySrcColumn(srcColumnList, lastColumnList, task.getMappingColumn());
    task.setDstColumnList(lastColumnList);

    ApiDefinitionDto queryApiDefinitionDto = new ApiDefinitionDto();
    queryApiDefinitionDto.setJobId(jobDTO.getId());
    ApiDefinitionDto apiDefinitionDto = apiDefinitionService.selectOne(queryApiDefinitionDto);
    if (apiDefinitionDto != null && StringUtils.isNotEmpty(apiDefinitionDto.getDefinitionJson())) {
      APIHttpDefinition apiHttpDefinition =
          JSON.parseObject(apiDefinitionDto.getDefinitionJson(), APIHttpDefinition.class);
      jobDTO.setApiHttpDefinition(apiHttpDefinition);

      JSONObject json = new JSONObject();
      json.put(Key.DATA_MODE, apiHttpDefinition.getDataMode());
      json.put(Key.DATA_ROOT_PATH, apiHttpDefinition.getDataRootPath());
      task.setDataxJson(json.toJSONString());
    }

    return task;
  }

  /** 更新任务 */
  @Override
  public String updateTask(TaskDTO oldTask, TaskDTO taskDTO) throws Exception {
    JobDTO jobDTO = jobService.selectByPrimaryKey(oldTask.getJobId());
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), oldTask.getJobId()));
    CheckParaUtils.checkStringValidity(taskDTO.getDstTableName(), logger, "目标表名必填");
    CheckParaUtils.checkObjectValidity(taskDTO.getModifyColumn(), logger, "是否修改列映射必填");

    // 记录日志
    BdpLogUtil.log4Update(taskDTO, oldTask);

    // 判断列名称合法性
    if (!StringUtils.isEmpty(taskDTO.getMappingColumn())) {
      List<ColumnMappingDto> columnMappingDtos =
          JSONUtils.toList(taskDTO.getMappingColumn(), ColumnMappingDto.class);
      this.checkColumnName(columnMappingDtos);
    }

    // 将数据根路径和源字段组合
    //        appendDataRootPath(taskDTO);

    // 更新数据结构和数据根路径
    ApiDefinitionDto queryApiDefinitionDto = new ApiDefinitionDto();
    queryApiDefinitionDto.setJobId(jobDTO.getId());
    ApiDefinitionDto apiDefinitionDto = apiDefinitionService.selectOne(queryApiDefinitionDto);
    CheckParaUtils.checkDbObjectValidity(
        apiDefinitionDto, logger, String.format("API定义数据没查到,jobId:%s", jobDTO.getId()));

    APIHttpDefinition apiHttpDefinition =
        JSON.parseObject(apiDefinitionDto.getDefinitionJson(), APIHttpDefinition.class);
    if (apiHttpDefinition == null) {
      throw new BusinessException("请检查任务定义API参数");
    }
    ApiDefinitionDto deleteDto = new ApiDefinitionDto();
    deleteDto.setId(apiDefinitionDto.getId());
    apiDefinitionService.delete(deleteDto);

    JSONObject json = JSONObject.parseObject(taskDTO.getDataxJson());
    apiHttpDefinition.setDataMode(json.getString(Key.DATA_MODE));
    apiHttpDefinition.setDataRootPath(json.getString(Key.DATA_ROOT_PATH));
    apiDefinitionDto.setDefinitionJson(JSONObject.toJSONString(apiHttpDefinition));
    apiDefinitionDto.setId(ComUtil.getId());
    apiDefinitionService.insert(apiDefinitionDto);

    // 先更新列映射关系，再生成dataxjson
    taskDTO.setJobId(oldTask.getJobId());
    taskService.updateByPrimaryKeySelective(taskDTO);
    return dataxJsonService.buildDataxJson(oldTask);
  }

  /**
   * 新增task
   *
   * @param jobDTO
   * @param tableMappingDto
   * @throws Exception
   */
  private TaskDTO insertTask(JobDTO jobDTO, TableMappingDto tableMappingDto) throws Exception {
    TaskDTO taskDTO = new TaskDTO();
    taskDTO.setId(ComUtil.getId());
    taskDTO.setJobId(jobDTO.getId());
    taskDTO.setJobType(jobDTO.getJobType());
    taskDTO.setSrcTableName(tableMappingDto.getSrc_table());
    taskDTO.setDstTableName(tableMappingDto.getDst_table());
    taskDTO.setMappingColumn(null);
    taskDTO.setSyncMode(jobDTO.getSyncMode());
    taskDTO.setIsCreated(Enable.ZERO.getCode());
    taskDTO.setFlag(Enable.ZERO.getCode());
    taskService.insert(taskDTO);

    // 记录日志
    BdpLogUtil.log4Insert(taskDTO);
    logger.info("新增任务成功,dstTable:{}", tableMappingDto.getDst_table());

    return taskDTO;
  }

  /**
   * 更新task
   *
   * @param jobDTO
   * @param tableMappingDto
   * @throws Exception
   */
  private TaskDTO updateTask(JobDTO jobDTO, TableMappingDto tableMappingDto) throws Exception {
    TaskDTO query = new TaskDTO();
    query.setJobId(jobDTO.getId());
    query.setFlag(Enable.ZERO.getCode());
    query.setSrcTableName(tableMappingDto.getSrc_table());
    TaskDTO taskDTO = taskService.selectOne(query);

    // 目标表发生改变需要更新数据
    if (!taskDTO.getDstTableName().equals(tableMappingDto.getDst_table())) {
      taskDTO.setMappingColumn(null);
      taskDTO.setDstTableName(tableMappingDto.getDst_table());
      taskDTO.setFilterCondition(null);
      taskDTO.setModifyColumn(Enable.ZERO.getCode());
      taskDTO.setUpdateHistoryData(Enable.ONE.getCode());
      taskDTO.setSyncConditionField(null);
      taskDTO.setFieldName(null);
      taskDTO.setSyncTimePoint(null);
      taskDTO.setSyncIdValue(null);
    }

    taskDTO.setSyncMode(jobDTO.getSyncMode());
    taskDTO.setIsCreated(Enable.ZERO.getCode());
    taskService.updateByPrimaryKey(taskDTO);
    logger.info("修改任务成功,dstTable:{}", tableMappingDto.getDst_table());

    // 删除上次任务保存bdp_dcs_task_table_detail表数据和保存bdp_dcs_task_data_detail表数据
    TaskTableDetailDTO taskTableDetailDTO =
        TaskTableDetailDTO.builder().taskId(taskDTO.getId()).build();
    taskTableDetailService.delete(taskTableDetailDTO);

    TaskDataDetailDTO taskDataDetailDTO =
        TaskDataDetailDTO.builder().taskId(taskDTO.getId()).build();
    taskDataDetailService.delete(taskDataDetailDTO);
    return taskDTO;
  }

  /**
   * 删除任务
   *
   * @param jobDTO
   * @param srcTableList
   * @throws Exception
   */
  private void deleteTask(JobDTO jobDTO, List<String> srcTableList) throws Exception {
    // 根据jobId查询taskList，和selectedTable比较，selectedTable里边不存在的srcTable，就删除task，同时删除垃圾数据
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);
    List<String> srcTableMappingList =
        tableMappingDtoList.stream()
            .map(TableMappingDto::getSrc_table)
            .collect(Collectors.toList());
    List<String> collect =
        srcTableList.stream()
            .filter(s -> !srcTableMappingList.contains(s))
            .collect(Collectors.toList());
    collect.forEach(
        srcTableName -> {
          TaskDTO taskDtoQuery = new TaskDTO();
          taskDtoQuery.setJobId(jobDTO.getId());
          taskDtoQuery.setSrcTableName(srcTableName);
          taskDtoQuery.setFlag(Enable.ZERO.getCode());
          List<TaskDTO> taskList = taskService.selectList(taskDtoQuery);
          taskList.forEach(
              task -> {
                TaskTableDetailDTO taskTableDetailDTO = new TaskTableDetailDTO();
                taskTableDetailDTO.setTaskId(task.getId());
                taskTableDetailService.delete(taskTableDetailDTO);
                TaskDataDetailDTO taskDataDetailDTO = new TaskDataDetailDTO();
                taskDataDetailDTO.setTaskId(task.getId());
                taskDataDetailService.delete(taskDataDetailDTO);

                // 删除任务
                task.setFlag(1);
                taskService.updateByPrimaryKey(task);
              });
          //            taskService.delete(taskDtoQuery);
          logger.info("删除任务成功,srcTableName:{}", srcTableName);
        });
  }

  /** 批量保存任务 一个job关联多个task */
  @Override
  public void saveTasksByJobId(JobDTO jobDTO) throws Exception {
    List<String> srcTableList = taskService.getSrcTableListByJobId(jobDTO.getId());

    // 根据jobId和srcTable查询task存不存在，存在就更新，不存在新增
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);

    for (TableMappingDto tableMappingDto : tableMappingDtoList) {
      TaskDTO taskDTO = null;
      if (!srcTableList.contains(tableMappingDto.getSrc_table())) {
        taskDTO = insertTask(jobDTO, tableMappingDto);
      } else {
        taskDTO = updateTask(jobDTO, tableMappingDto);
      }

      // 保存bdp_dcs_task_table_detail表数据和保存bdp_dcs_task_data_detail表数据
      taskService.saveTaskDetailData(
          tableMappingDto.getSrc_table(), tableMappingDto.getDst_table(), taskDTO, jobDTO);
    }

    // 删除任务
    deleteTask(jobDTO, srcTableList);
  }

  /**
   * 创建目标表并回写建表sql语句
   *
   * @param jobDTO jobDTO
   * @param taskDTO taskDTO
   * @return 创建目标表结果
   * @throws Exception
   */
  @Override
  public Boolean createTable(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    logger.info("processMode:{}", jobDTO.getProcessMode());
    boolean createTable = createTablePreHandle(jobDTO, taskDTO);

    if (!createTable) {
      return true;
    }

    Map<String, String> map = this.createDstTable(jobDTO, taskDTO);
    return updateSql(map, taskDTO);
  }

  /** 创建表 */
  public Map<String, String> createDstTable(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    String srcTable = taskDTO.getSrcTableName();
    String dstTable = taskDTO.getDstTableName();
    DataSourceDTO srcDataSource = dataSourceUtils.queryDataSourceById(jobDTO);
    DataSourceDTO dstDataSource =
        dataSourceService.queryDataSourceById(jobDTO.getDstDatasourceId());
    List<ColumnMeta> columnList = new ArrayList<>();

    // 获取接口字段及类型
    List<ColumnMappingDto> columnMappingList =
        JSONUtils.toList(taskDTO.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingList)) {
      for (ColumnMappingDto columnMapping : columnMappingList) {
        ColumnMeta columnMeta = new ColumnMeta();
        columnMeta.setName(columnMapping.getDstColumnName());
        columnMeta.setColumnType(columnMapping.getSrcColumnType());
        columnMeta.setNullable(1);
        columnList.add(columnMeta);
      }
    }
    logger.info("jobId={},源数据表:{},字段为:{}", jobDTO.getId(), srcTable, columnList);

    Map<String, Object> paramMap = new HashMap<>();
    return DBUtils.createTable(srcDataSource, dstDataSource, columnList, null, dstTable, paramMap);
  }

  private Boolean createTablePreHandle(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    boolean existDstTable =
        taskService.isExistTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
    logger.info("existDstTable:{}", existDstTable);
    taskService.saveTableLog(taskDTO, String.format("目标库存在目标表:%s", existDstTable));

    // 目标表不存在，直接新建表
    if (existDstTable) {
      // 目标已存在表的处理模式，预检查并报错拦截：0，覆盖目标表：1，保留目标表：2
      if (jobDTO.getProcessMode() == ProcessMode.RETAIN_DST_TABLE.getCode()) {
        logger.info("已存在表的处理模式为保留目标表，不做处理");
        return false;
      } else if (jobDTO.getProcessMode() == ProcessMode.CLEAR_DST_TABLE_DATA.getCode()) {
        boolean isClear =
            baseDbService.clearTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
        logger.info("清空目标表数据:{},清空结果:{}", taskDTO.getDstTableName(), isClear);
        return false;
      } else {
        throw new BusinessException("不支持的已存在表的处理模式");
      }
    }

    return true;
  }

  /** 回写创建表sql语句 */
  private Boolean updateSql(Map<String, String> map, TaskDTO taskDTO) throws Exception {
    taskService.saveTableLog(
        taskDTO, String.format("创建表sql:%s", map.get(Constants.CREATE_TABLE_SQL)));
    boolean createTableFlag = true;

    TaskTableDetailDTO query =
        TaskTableDetailDTO.builder()
            .taskId(taskDTO.getId())
            .srcTableName(taskDTO.getSrcTableName())
            .build();
    TaskTableDetailDTO tableDetailDTO = taskTableDetailService.selectOne(query);
    CheckParaUtils.checkDbObjectValidity(
        tableDetailDTO, logger, String.format("任务id:%s,没找到对应的表结构迁移明细数据", taskDTO.getId()));

    tableDetailDTO.setCreateSql(map.get(DcsConstants.CREATE_TABLE_SQL));
    if (Boolean.parseBoolean(map.get(Constants.SUCCESS))) {
      tableDetailDTO.setTaskStatus(TaskDetailStatus.COMPLETED.getCode());

      taskDTO.setIsCreated(Enable.ONE.getCode());
      taskService.updateByPrimaryKeySelective(taskDTO);
      logger.info("表:{} 更新为已创建", taskDTO.getDstTableName());

      taskService.saveTableLog(taskDTO, String.format("创建表成功"));
    } else {
      tableDetailDTO.setTaskStatus(TaskDetailStatus.EXECUTE_FAIL.getCode());
      createTableFlag = false;

      taskService.saveTableLog(taskDTO, String.format("创建表失败,原因是:%s", map.get(Constants.MESSAGE)));
    }
    taskTableDetailService.updateByPrimaryKeySelective(tableDetailDTO);
    logger.info("创建表sql={}", map.get(DcsConstants.CREATE_TABLE_SQL));
    logger.info(
        "************创建表返回结果：{},状态为:{},表名:{}",
        map.get(Constants.SUCCESS),
        tableDetailDTO.getTaskStatus(),
        taskDTO.getDstTableName());
    return createTableFlag;
  }

  /** 检查列名合法性 */
  private void checkColumnName(List<ColumnMappingDto> columnMappingDtos) {
    for (ColumnMappingDto columnMappingDto : columnMappingDtos) {
      if (!tableService.isValidColumnName(columnMappingDto.getDstColumnName())) {
        throw new BusinessException(
                String.format(
                        DcsErrorCode.COLUMN_NAME_CHECK.getDesc(), columnMappingDto.getDstColumnName()));
      }

      // 字段类型不能是number
      if("number".equalsIgnoreCase(columnMappingDto.getSrcColumnType())){
        throw new BusinessException("字段名:["+columnMappingDto.getSrcColumnName()+"],数据类型不能为number");
      }
    }
  }

  @Override
  public Integer getType() {
    return JobType.API_HTTP.getCode();
  }

}
